package edu.ucla.cs.rpc.multicast.network.message;

import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.net.Socket;
import java.net.SocketAddress;

/**
 * A Message can be sent point-to-point over the network. It has a type, a
 * source socket address, and optionally, a sender ID, sequence number, and
 * timestamp.
 * 
 * @author Chase Covello Philip Russell
 * 
 */
public class Message implements Serializable {

	/**
	 * Message types that can be assigned to messages.
	 * 
	 * @author Chase Covello Philip Russell
	 * 
	 */
	public static enum MessageType {
		JOIN_RECEIVERS, JOIN_SENDERS, LEAVE_RECEIVERS, LEAVE_SENDERS, RPC_MESSAGE;
	}

	/**
	 * Constant representing no timestamp.
	 */
	public static final long NO_TIMESTAMP = -1;

	private static final long serialVersionUID = 6L;

	private long senderID;

	private long seqNum;

	private final SocketAddress source;

	private long timestamp;

	private final MessageType type;

	/**
	 * Construct a new message using information from the given message.
	 * 
	 * @param m
	 *            the message to copy.
	 */
	public Message(Message m) {
		this.type = m.type;
		this.source = m.source;

		this.seqNum = m.seqNum;
		this.senderID = m.senderID;
		this.timestamp = m.timestamp;
	}

	/**
	 * Construct a new message with the given type and source.
	 * 
	 * @param type
	 * @param source
	 */
	public Message(MessageType type, SocketAddress source) {
		this.type = type;
		this.source = source;
	}

	/**
	 * Compares the given Message to this Message. The two are equal if their
	 * hash codes are equal.
	 * 
	 * @param m
	 *            the message to compare to.
	 * @return true if the two hash codes are equal.
	 */
	public boolean equals(Message m) {
		return hashCode() == m.hashCode();
	}

	/**
	 * Returns this Message's sender ID.
	 * 
	 * @return the sender ID.
	 */
	public long getSenderID() {
		return senderID;
	}

	/**
	 * Returns this Message's sequence number.
	 * 
	 * @return the sequence number.
	 */
	public long getSeqNum() {
		return seqNum;
	}

	/**
	 * Returns this Message's source socket address.
	 * 
	 * @return the socket address.
	 */
	public SocketAddress getSource() {
		return source;
	}

	/**
	 * Returns this Message's timestamp.
	 * 
	 * @return the timestamp.
	 */
	public long getTimestamp() {
		return timestamp;
	}

	/**
	 * Returns this Message's type.
	 * 
	 * @return the message type.
	 */
	public MessageType getType() {
		return type;
	}

	/**
	 * Returns a hash code for this Message. The hash code is given by the
	 * formula <code>type.hashCode() ^ source.hashCode()</code>.
	 */
	public int hashCode() {
		return type.hashCode() ^ source.hashCode();
	}

	/**
	 * Send this message to the given destination. This method guarantees
	 * reliable, in-order delivery of messages.
	 * 
	 * @param dest
	 *            the address of the message's destination.
	 * @throws IOException
	 *             in the case of unrecoverable network errors.
	 */
	public void send(SocketAddress dest) throws IOException {
		Socket socket = new Socket();
		socket.connect(dest);

		ObjectOutputStream oos = new ObjectOutputStream(socket
				.getOutputStream());
		oos.writeObject(this);

		socket.close();
	}

	/**
	 * Sets this Message's sender ID to the given ID.
	 * 
	 * @param senderID
	 *            the sender ID.
	 */
	public void setSenderID(long senderID) {
		this.senderID = senderID;
	}

	/**
	 * Sets this Message's sequence number to the given sequence number.
	 * 
	 * @param seqNum
	 *            the sequence number.
	 */
	public void setSeqNum(long seqNum) {
		this.seqNum = seqNum;
	}

	/**
	 * Sets this Message's timestamp to the given timestamp.
	 * 
	 * @param timestamp
	 *            the timestamp.
	 */
	public void setTimestamp(long timestamp) {
		this.timestamp = timestamp;
	}

	/**
	 * Returns a string representation of this Message.
	 */
	public String toString() {
		return getClass().getSimpleName() + " type=" + type + ", source="
				+ source + ", seqNum=" + seqNum + ", senderID=" + senderID
				+ ", timestamp=" + timestamp;
	}
}
